/*
 * Icegem, Extensions library for VMWare vFabric GemFire
 * 
 * Copyright (c) 2010-2011, Grid Dynamics Consulting Services Inc. or third-party  
 * contributors as indicated by the @author tags or express copyright attribution
 * statements applied by the authors.  
 * 
 * This copyrighted material is made available to anyone wishing to use, modify,
 * copy, or redistribute it subject to the terms and conditions of the GNU
 * Lesser General Public License v3, as published by the Free Software Foundation.
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
 * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 * 
 * You should have received a copy of the GNU Lesser General Public License v3
 * along with this distribution; if not, write to:
 * Free Software Foundation, Inc.
 * 51 Franklin Street, Fifth Floor
 * Boston, MA  02110-1301  USA
 */
package com.googlecode.icegem.cacheutils.regioncomparator;

import com.gemstone.gemfire.cache.CacheFactory;
import com.gemstone.gemfire.cache.Region;
import com.gemstone.gemfire.cache.execute.FunctionAdapter;
import com.gemstone.gemfire.cache.execute.FunctionContext;
import com.gemstone.gemfire.cache.partition.PartitionRegionHelper;
import com.gemstone.gemfire.internal.cache.PartitionedRegion;

import java.util.HashSet;
import java.util.Set;

/**
 * execute on partitioned region, to retrieve bucket keys
 *
 * first param is region name
 * second param is bucket id
 *
 * return keys in this bucket
 *
 * User: Artem Kondratyev, e-mail: kondratevae@gmail.com
 */
public class BucketKeysGetterFunction extends FunctionAdapter{

    @Override
    public void execute(FunctionContext functionContext) {
        String regionName = (String) ((Object[]) functionContext.getArguments())[0];
        Integer bucketId = (Integer) ((Object[]) functionContext.getArguments())[1];
        Region region = CacheFactory.getAnyInstance().getRegion(regionName);
        //todo: remove exceptions, unnecessary. skip member
        if (region == null)
            throw  new RuntimeException("there's no such region \'" + regionName + "\' on this sever");
        if (!PartitionRegionHelper.isPartitionedRegion(region))
            throw  new RuntimeException("this region \'" + regionName + "\' is not partitioned");

        Set keys = new HashSet();
        try {
            PartitionedRegion pr = (PartitionedRegion)region;
            if (pr.getDataStore().getAllLocalPrimaryBucketIds().contains(bucketId))
                keys.addAll(pr.getDataStore().getKeysLocally(bucketId));
        } catch (Exception e) {
            throw new RuntimeException("error getting keys for bucket " + bucketId, e);
        }

        functionContext.getResultSender().lastResult((HashSet) keys);
    }

    @Override
    public String getId() {
        return getClass().getName();
    }
}
